package com.erik.lin.util;

import com.erik.lin.annotion.MessageListener;
import com.google.common.eventbus.AsyncEventBus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

/**
 * Created by Erik on 2017/4/5.
 */
public class MqUtil {

    private Logger LOG = LoggerFactory.getLogger(MqUtil.class);
    private static MqUtil instance = new MqUtil();
    private Map<String, AsyncEventBus> queues = new ConcurrentHashMap();

    public static MqUtil getInstance() {
        return instance;
    }

    private MqUtil() {
        PathMatchingResourcePatternResolver resourcePatternResolver = new PathMatchingResourcePatternResolver();
        String basePkg = "com/erikhenry/lin";
        String packageSearchPath = "classpath*:" + basePkg + "/**/mq/listener/*.class";

        try {
            Resource[] e = resourcePatternResolver.getResources(packageSearchPath);
            Resource[] var8 = e;
            int var7 = e.length;

            for(int var6 = 0; var6 < var7; ++var6) {
                Resource resource = var8[var6];
                String uri = resource.getURI().toString();
                String className = uri.substring(uri.indexOf(basePkg), uri.length() - 6).replace('/', '.');
                Class cls = Class.forName(className);
                if(cls.isAnnotationPresent(MessageListener.class)) {
                    String queueName = ((MessageListener)cls.getAnnotation(MessageListener.class)).queueName()[0];
                    LOG.info("监听队列{},{}", queueName, className);
                    AsyncEventBus queue = this.createQueue(queueName);
                    queue.register(cls.newInstance());
                }
            }
        } catch (IOException var14) {
            var14.printStackTrace();
        } catch (ClassNotFoundException var15) {
            var15.printStackTrace();
        } catch (InstantiationException var16) {
            var16.printStackTrace();
        } catch (IllegalAccessException var17) {
            var17.printStackTrace();
        }
    }

    public AsyncEventBus createQueue(String queueName) {
        if(!this.queues.containsKey(queueName)) {
            AsyncEventBus queue = new AsyncEventBus(queueName, Executors.newFixedThreadPool(200));
            this.queues.put(queueName, queue);
        } else {
            LOG.warn("队列已存在");
        }

        return (AsyncEventBus)this.queues.get(queueName);
    }

    public void senderMessage(String destination, Object message) throws RuntimeException {
        if(this.queues.containsKey(destination)) {
            ((AsyncEventBus)this.queues.get(destination)).post(message);
        } else {
            LOG.warn("{}队列不存在", destination);
            throw new RuntimeException(destination + "队列不存在");
        }
    }

}
